/*
 * Copyright Hundsun Technologies Inc. All Rights Reserved.
 */

package checker

import (
	"errors"
	"fmt"
	"strconv"
	"sync"
	"time"

	"hundsun.com/hsl/hschain/consensus/types"

	"github.com/allegro/bigcache"
	"github.com/golang/protobuf/proto"
	"go.uber.org/atomic"
	"hundsun.com/hsl/hschain/common/log/logging"
	"hundsun.com/hsl/hschain/common/util/pb"
	dposcomm "hundsun.com/hsl/hschain/consensus/impl/dpos/common"
	"hundsun.com/hsl/hschain/protos/consensus/dpos"
	pbledger "hundsun.com/hsl/hschain/protos/ledger"
)

var roundMinersCache *bigcache.BigCache
var checkerOnce sync.Once

// Checker 拉取块并负责验证
type Checker struct {
	module                       *DposChecker
	lastAcceptedBlockRound       int64
	lastAcceptedBlockRoundMiners *dposcomm.SharedMiners
}

// Init 初始化
func (ck *Checker) Init(blockCheckInterval, pendingBlockPoolFetchSize int) error {
	var err error
	checkerOnce.Do(func() {
		config := bigcache.Config{
			Shards:             16,
			LifeWindow:         time.Duration(120) * time.Second,
			CleanWindow:        time.Duration(5) * time.Second,
			MaxEntriesInWindow: 100,
			Logger:             ck.module.Log,
		}
		roundMinersCache, err = bigcache.NewBigCache(config)
		if err != nil {
			ck.module.Log.Errorf("failed to create BigCache instance when initializing BlockExecutedMessage cache: %s", err.Error())
		}
	})
	blockAcceptedCallback := func(blk *pbledger.Block) {
		ck.lastAcceptedBlockRound, _, _ = dposcomm.CalculateRound(blk.Header.BlockTime)
	}
	adapterConf := &types.CheckerConfig{
		ModuleLog:                 ck.module.Log,
		BlockCheckInterval:        blockCheckInterval,
		BlockFetchInterval:        dposcomm.TimeSliceLength * 1000,
		PendingBlockPoolFetchSize: pendingBlockPoolFetchSize,
		BlockCreatorValidator:     ck.createBlockCreatorValidator(),
		BlockAcceptedCallback:     blockAcceptedCallback,
	}
	return ck.module.Client.InitChecker(adapterConf)
}

func (ck *Checker) createBlockCreatorValidator() types.BlockCreatorValidator {
	return func(blk *pbledger.Block) (err error) {
		log := pb.GetBlockLog(ck.module.Log, blk)
		blkCreator := blk.Header.Signature.GetAddress()

		// 校验区块的序列化签名
		err = dposcomm.ValidateSequenceSignature(blk, ck.module.Log)
		if err != nil {
			return err
		}
		log.Debug("block sequence signature validate passed.")

		var genesisMiners = &dposcomm.SharedMiners{}
		genesisMiners.ReplaceAll(ck.module.Client.GetCurrentMiners())

		//先验证块是否由genesisMiners出的
		var genesisChkErr error
		creatorIndexOfGenesisMiners := genesisMiners.IndexOf(blkCreator)
		if creatorIndexOfGenesisMiners >= 0 {
			t1 := time.Now()
			genesisChkErr = validateBlockCreatorAndEndorsers(blk, genesisMiners, log)
			log.WithField(logging.LogFieldCost, time.Since(t1)/time.Millisecond).Debugf("validate with genesis miners,err:%v", genesisChkErr)
		}
		if genesisChkErr != nil || creatorIndexOfGenesisMiners < 0 {
			blkRound, roundStartUnixMilli, _ := dposcomm.CalculateRound(blk.Header.BlockTime)
			roundGap := blkRound - ck.lastAcceptedBlockRound
			if roundGap > 1 || roundGap < 0 {
				if genesisChkErr != nil {
					return errors.New("failed to check block with genesis miners when round gap is greater than one")
				}
				return errors.New("block is not created by genesis miners when round gap is greater than one or less than zero")
			}

			t2 := time.Now()
			var blkRoundMiners *dposcomm.SharedMiners
			if roundGap == 0 {
				blkRoundMiners = ck.lastAcceptedBlockRoundMiners
			} else {
				//roundGap=1时必须等到等到区块执行完毕
				for blkRoundMiners == nil {
					blkRoundMiners, _ = ck.getMinersOfRoundFromStorage(blkRound, log)
					time.Sleep(time.Duration(10) * time.Millisecond)
				}
			}
			log.WithField(logging.LogFieldCost, time.Since(t2)/time.Millisecond).Debug("got miners of block round")

			creatorIndexBlkRoundMiners := blkRoundMiners.IndexOf(blkCreator)
			if creatorIndexBlkRoundMiners < 0 {
				return fmt.Errorf("creator '%s'is not a miner of that round", blkCreator)
			}
			err = checkTimeSlice(roundStartUnixMilli, int64(creatorIndexBlkRoundMiners), blk.Header.BlockTime)
			if err != nil {
				return err
			}

			t3 := time.Now()
			err = validateBlockCreatorAndEndorsers(blk, blkRoundMiners, log)
			log.WithField(logging.LogFieldCost, time.Since(t3)/time.Millisecond).Debug("block validated")
			if err != nil {
				return err
			}
		}
		return nil
	}
}

func (ck *Checker) getMinersOfRoundFromStorage(round int64, log logging.Log) (*dposcomm.SharedMiners, error) {
	var roundMiners dpos.RoundMiners
	cacheKey := strconv.FormatInt(round, 10)
	roundBytes, err := roundMinersCache.Get(cacheKey)
	if err == nil {
		_ = proto.Unmarshal(roundBytes, &roundMiners)
		log.Debugf("miners of round[%d] cache hit", round)
	} else {
		log.Debugf("fetching miners of round[%d] from storage", round)
		if roundBytes, err = ck.module.Client.RequestRoundMinersInfo(round, types.DefaultMqTimeOut); err != nil {
			log.Warnf("failed to get miners of round[%d] from storage", round)
			return nil, err
		}
		log.Debugf("got miners of round[%d] from storage", round)
		err = roundMinersCache.Set(cacheKey, roundBytes)
		if err != nil {
			log.Warnf("failed to cache miners of round[%d] ", round)
		} else {
			log.Debugf("miners of round[%d] cached", round)
		}
		_ = proto.Unmarshal(roundBytes, &roundMiners)
	}

	if len(roundMiners.Miners) > 0 {
		miners := &dposcomm.SharedMiners{}
		for _, m := range roundMiners.Miners {
			miners.Add(m)
		}
		return miners, nil
	}

	return nil, nil
}

// validateBlockCreatorAndEndorsers 验证区块创建者是本轮矿工， 背书者是否是矿工，背书签名是否正确（背书有14个正确即可）
func validateBlockCreatorAndEndorsers(block *pbledger.Block, miners *dposcomm.SharedMiners, log logging.Log) error {
	//1.校验区块签名基本格式和背书个数
	if block.Header.Signature == nil || len(block.Header.Signature.Signature) == 0 || len(block.Header.Signature.PublicKey) == 0 {
		return errors.New("signature content format error")
	}

	consensusBytes := block.Header.Consensus
	if len(consensusBytes) == 0 {
		return errors.New("'consensus' field in block header is empty")
	}
	var consensus dpos.BlockEndorsementResponses
	err := proto.Unmarshal(consensusBytes, &consensus)
	if err != nil {
		return errors.New("unable to parse 'consensus' field in block header")
	}
	if len(consensus.Responses) < dposcomm.MinApproveMinerCount {
		return errors.New("count of endorsements is not enough")
	}

	//2.校验块签名和背书签名
	blockCreator := block.Header.Signature.GetAddress()
	minerIndex := miners.IndexOf(blockCreator)
	if minerIndex < 0 {
		return fmt.Errorf("creator '%s' of block[%d] is not a miner of that round", blockCreator, block.Header.Number)
	}
	//3.验证背书签名
	validEdCount := atomic.NewInt32(0)
	blockSigValid := atomic.NewBool(false)
	var wg sync.WaitGroup
	wg.Add(len(consensus.Responses) + 1)
	go func() {
		blockSigValid.Store(block.Verify())
		wg.Done()
	}()
	for i := 0; i < len(consensus.Responses); i++ {
		go func(j int) {
			ed := consensus.Responses[j]
			edAddr := ed.GetSignature().GetAddress()
			if edAddr != blockCreator && miners.IndexOf(edAddr) >= 0 && ed.Verify() {
				validEdCount.Add(1)
			}
			wg.Done()
		}(i)
	}
	wg.Wait()
	log.Debug("block signature and ed signature validated")
	if !blockSigValid.Load() {
		return fmt.Errorf("failed to verify signature fo block '%d'", block.Header.Number)
	}

	if validEdCount.Load() < int32(dposcomm.MinApproveMinerCount) {
		return errors.New("count of valid endorsements is not enough")
	}
	return nil

}

func checkTimeSlice(roundStartUnixMilli, creatorIndex, blockTime int64) error {
	start := roundStartUnixMilli + int64(dposcomm.TimeSliceLength)*1000*creatorIndex
	end := roundStartUnixMilli + int64(dposcomm.TimeSliceLength)*1000*(creatorIndex+1)
	if blockTime < start || blockTime >= end {
		return fmt.Errorf("blockTime of block[%d] is not in creator's time slice", blockTime)
	}
	return nil
}
